package flink_p1

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, createTypeInformation}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 *
 * 间歇性watermark
 *
 *
 * */

object FlinkTest_20_window_watermark_PunctuatedWatermarks {


  /**
   * 使用event time来计算WordCount
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val socketStream: DataStream[String] = env.socketTextStream("node1", 8889)
      .assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks[String] {
        var maxTimeStamp: Long = _

        override def checkAndGetNextWatermark(lastElement: String, extractedTimestamp: Long): Watermark = {

          println(s"extractedTimestamp: $extractedTimestamp")
          maxTimeStamp = extractedTimestamp.max(maxTimeStamp)

          //仅针对数据为a
          if ("a".equals(lastElement.split(" ")(1))) {
            new Watermark(maxTimeStamp - 2000)
          } else {
            new Watermark(maxTimeStamp)
          }
        }

        override def extractTimestamp(element: String, previousElementTimestamp: Long): Long = {
          element.split(" ")(0).toLong
        }
      })

    socketStream.map(data => {
      val arr: Array[String] = data.split(" ")
      (arr(1), 1)
    }).keyBy(_._1)
      .timeWindow(Time.seconds(3), Time.seconds(3))
      .reduce(new ReduceFunction[(String, Int)] {
        override def reduce(value1: (String, Int), value2: (String, Int)): (String, Int) = {
          (value1._1, value1._2 + value2._2)
        }
      }, new WindowFunction[(String, Int), (String, Int), String, TimeWindow] {
        override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {
          println(s"win ${window.getStart}    ${window.getEnd}")
          for (elem <- input) {
            out.collect(elem)
          }
        }
      }).print()

    env.execute()
  }

}
